package com.leiyuee.flink.tools;

import com.alibaba.fastjson.JSONObject;
import com.leiyuee.flink.tools.enums.RowDataFuncEnums;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
import org.apache.kafka.clients.consumer.ConsumerRecord;


import java.nio.charset.StandardCharsets;
import java.util.function.Function;

public class CustomKafkaDeserialization implements KafkaDeserializationSchema<RowData> {

    private CustomKafkaDeserialization() {}

    //private String enumIndex;
    private RowDataFuncEnums enumName;
    public CustomKafkaDeserialization(RowDataFuncEnums enumName) {
        this.enumName = enumName;
    }


    //public CustomKafkaDeserialization(String enumIndex, RowDataFuncEnums enumName) {
    //    this.enumIndex = enumIndex;
    //    this.enumName = enumName;
    //}

    @Override
    public RowData deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        if(record != null && record.value() != null) {
            String valueStr = new String(record.value(), StandardCharsets.UTF_8);
            // 将 value 的数据获取出来，源数据
            final JSONObject valueJson = JSONObject.parseObject(valueStr);
            RowKind rowKind = RowKind.INSERT;
            // 操作数据
            JSONObject operateJson = valueJson;

            if (valueJson.containsKey("op")) {
                String type = valueJson.get("op").toString().toLowerCase();
                if (type.equals("d") || type.equals("delete")) {
                    rowKind = RowKind.DELETE;
                    operateJson = valueJson.getJSONObject("before");
                } else if (type.equals("u") || type.equals("update")) {
                    rowKind = RowKind.UPDATE_AFTER;
                    operateJson = valueJson.getJSONObject("after");
                } else {
                    operateJson = valueJson.getJSONObject("after");
                }
            }

            // 注意：构造传参无法将 Function 序列化，所以此处采用枚举的形式去获取对应的 Function，Function 的配置则在 RowDataFuncEnums 里面，可以自定义一些常量类来配置 Function

            // 方法①：使用构造函数保存一个 Enum 的标识符获取 Function
            //Function<JSONObject, RowData> function = RowDataFuncEnums.getFuncById(this.enumIndex);

            // 方法②：使用构造函数保存一个 Enum 获取 Function
            RowDataFuncEnums funcEnums = this.enumName;
            Function<JSONObject, RowData> function = funcEnums.getFunction();
            if (function == null) {
                return null;
            }

            RowData rowData = function.apply(operateJson);
            rowData.setRowKind(rowKind);
            return rowData;
        }
        return null;
    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return TypeInformation.of(RowData.class);
    }
}
